package com.shujia.windowApply;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class SessionWindowApply {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> linesDS = env.socketTextStream("hadoop102", 8888);

        //解析数据
        DataStream<Tuple2<String, Long>> wordsDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        //指定时间字段和水位线
        DataStream<Tuple2<String, Long>> assDS = wordsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        //指定水位线，水位线等于最新一条数据的时间戳
                        .<Tuple2<String, Long>>forMonotonousTimestamps()
                        //指定时间字段
                        .withTimestampAssigner((kv, ts) -> kv.f1)
        );

        DataStream<Tuple2<String, Integer>> kvDS = assDS
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));

        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);


        /*
         * 会话窗口：如果一个key一段时间没有数据生成一个窗口进行计算
         * ProcessingTimeSessionWindows：处理时间的会话窗口
         * EventTimeSessionWindows： 事件时间的会话窗口
         *
         */
        keyByDS.window(EventTimeSessionWindows.withGap(Time.seconds(10)))
                .sum(1)
                .print();

        env.execute();

    }
}
